package lzzsrj.union.message.redis;

import com.alibaba.fastjson.JSON;
import lombok.SneakyThrows;
import lzzsrj.union.message.common.bean.UnionMessage;
import lzzsrj.union.message.redis.listener.RedisMQListenerMethod;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

@Component
public class RedisMQMessageRegister implements ApplicationRunner {

    @Autowired
    private RedissonClient redissonClient;

    List<RedisMQListenerMethod> redisMQListenerMethodList = new ArrayList<>();

    @Override
    public void run(ApplicationArguments args) throws Exception {

        this.redisMQListenerMethodList = RedisListenerScanAnnotationProcessor.getRedisMQListenerMethodList();

        for (RedisMQListenerMethod redisMQListenerMethod : redisMQListenerMethodList) {
            new Thread(new Worker(
                    redisMQListenerMethod.getTopic(),
                    redisMQListenerMethod.getInterval(),
                    redisMQListenerMethod.getQuantity())).start();
        }
    }

    private class Worker implements Runnable {

        private String topic = "";
        private Long time;
        private Integer quantity;

        @SneakyThrows
        @Override
        public void run() {

//            RDelayedQueue<String> rDelayedQueue = null;

            RBlockingDeque<String> rBlockingDeque = redissonClient.getBlockingDeque(topic);
//            if (rBlockingDeque == null) {
//                return null;
//            }
//            rDelayedQueue = redissonClient.getDelayedQueue(rBlockingDeque);

            while (true) {

                String jsonObject = rBlockingDeque.take();
                System.out.println("--> 延迟队列获取数据:{}" + jsonObject);
                UnionMessage message = JSON.parseObject(jsonObject, UnionMessage.class);
                for (RedisMQListenerMethod redisMQListenerMethod : redisMQListenerMethodList) {
                    if (redisMQListenerMethod.getTopic().equals(topic)) {
                        redisMQListenerMethod.getTargetMethod().invoke(redisMQListenerMethod.getBean(), message);
                    }
                }

//                List<String> msg = multiRPopPipeline(queueName, quantity);
//                List<RedisMQMessage> redisMQMessages = new ArrayList<>();
//                for (String s : msg) {
//                    redisMQMessages.add(JSON.parseObject(s, RedisMQMessage.class));
//                }
//                if (redisMQMessages.size() > 0) {
//                    for (RedisMQListenerMethod redisMQListenerMethod : redisMQListenerMethodList) {
//                        if (redisMQListenerMethod.getQueue().equals(queueName)) {
//                            for (RedisMQMessage redisMQMessage : redisMQMessages) {
//                                redisMQListenerMethod.getTargetMethod().invoke(redisMQListenerMethod.getBean(), redisMQMessage);
//                            }
//                        }
//                    }
//                }
//                Thread.sleep(time);
            }
        }

        public Worker(String topic, Long time, Integer quantity) {
            this.topic = topic;
            this.time = time;
            this.quantity = quantity;
        }
    }

    /**
     * 一次性pop出指定数量的数据
     *
     * @param key  键
     * @param size 需要取出的元素个数
     * @return 返回取出的元素集合
     */
//    public List<String> multiRPopPipeline(String key, int size) {
//        // 获取当前队列里的值
//        int curSize = Math.toIntExact(redisTemplate.opsForList().size(key));
//        if (curSize == 0) {
//            return Collections.emptyList();
//        }
//        // 判断操作次数
//        List<String> collect = redisTemplate.executePipelined(new SessionCallback<String>() {
//            @Override
//            public String execute(RedisOperations redisOperations) throws DataAccessException {
//                final int finalSize = Math.toIntExact(Math.min(curSize, size));
//                for (int i = 0; i < finalSize; i++) {
//                    redisOperations.opsForList().rightPop(key);
//                }
//                return null;
//            }
//        }).stream().map(obj -> (String) obj).collect(Collectors.toList());
//        return collect;
//    }
}
